package cgl.sensorgrid.sopac.gps.filters;

import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.util.LinkedList;
import java.util.List;
import java.util.StringTokenizer;

import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.dom4j.io.OutputFormat;
import org.dom4j.io.SAXReader;
import org.dom4j.io.XMLWriter;

import cgl.narada.event.NBEvent;
import cgl.sensorgrid.filters.Filter;
import cgl.sensorgrid.sopac.gps.filters.RealtimeRdahmmRunner.RealtimeRdahmmSite;

/**
 * class for receiving state change information of real-time RDAHMM, and saving the results in an xml file
 * @author gaoxm
 */
public class RealtimeRdahmmResulter extends Filter {
	
	/**
	 * for processing messages received from NaradaBrokering
	 * @author gaoxm
	 */
	protected class MessageProcessor extends Thread {
		public void run() {
			boolean running = true;
			try {
				while (running) {
					synchronized (listCV) {
						if (eventList.size() == 0) {
							listCV.wait();
						}
						NBEvent nbEvent = (NBEvent) eventList.remove(0);
						byte[] data2 = nbEvent.getContentPayload();
						if (data2 != null && data2.length > 0) {
							processMessage(new String(data2));
						}
					}
				}
				return;
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}
		
		/**
		 * get the state change information from a message,
		 * @param msg a message is like "raap;2008-02-21_14:33:04;2to3"
		 */
		public void processMessage(String msg) {
			try {
				StringTokenizer st = new StringTokenizer(msg, "\n");
				while (st.hasMoreTokens()) {
					String token = st.nextToken();
					if (!token.startsWith("EOF")) {
						if (token.length() > 4) {
							UtilSet.log(1, "messsage received:" + token);
							String stationName = token.substring(0, 4);
							int scPos = token.indexOf(';', 5);
							String time = token.substring(5, scPos);
							String change = token.substring(scPos+1, token.length());
							addStateChange(stationName, time, change);							
						}
					}
				}
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}
	}
	

	/**
	 * the file path to the xml result file of real-time RDAHMM
	 */
	String xmlResPath;
	
	/**
	 * the xml document object for creating and modifying the xml result
	 */
	Document xmlResDoc;
	
	/**
	 * buffer of NaradaBrokering events
	 */
	LinkedList eventList;
	
	/**
	 * conditional variable for accessing eventList
	 */
	Object listCV;
	
	/**
	 * for processing NaradaBrokering messages
	 */
	MessageProcessor msgProcessor;
	
	/**
	 * the millisecond time when the xml document were saved last time; we save it every 10 minutes
	 */
	long lastSaveTime;
	
	/**
	 * the millisecond time when a new xml document was created : we create a new xml everyday
	 */
	long lastCreatTime;
	
	/**
	 * time interval to save the xml document: 10 minutes
	 */
	static long SAVE_INTERVAL = 10*60*1000;
	
	/**
	 * time interval to create a new xml document: 10 minutes
	 */
	static long CREATE_INTERVAL = 86400*1000;
	
	/** 
	 * @param commType	communication type of NaradaBrokering 
	 * @param nbHost	host name of the broker to connect to
	 * @param nbPort	port of the broker to connect to
	 * @param subTopic	topic to subscribe
	 * @param pubTopic	topic to publish
	 * @param xmlResPath the file path to the xml result file of real-time RDAHMM
	 */	
	public RealtimeRdahmmResulter(String commType, String nbHost, String nbPort, String subTopic, String pubTopic, String xmlResPath) {
		// TODO Auto-generated constructor stub
		this.xmlResPath = xmlResPath;
		eventList = new LinkedList();
		listCV = new Object();
		File xmlFile = new File(xmlResPath);
		if (xmlFile.exists()) {
			SAXReader xmlReader = new SAXReader();
			try {
				FileReader fr = new FileReader(xmlFile);
				xmlResDoc = xmlReader.read(fr);
				fr.close();							
			} catch (Exception ex) {
				ex.printStackTrace();
				UtilSet.log(0, "Fail to read the xml result file, about to exit...");
				System.exit(-1);
			}
		} else {
			xmlResDoc = DocumentHelper.createDocument();
			Element eleXml = xmlResDoc.addElement("xml");
		}
		lastCreatTime = lastSaveTime = System.currentTimeMillis();
		
		String[] initArgs = {commType, nbHost, nbPort, subTopic, pubTopic};		
		initialize(initArgs);
		msgProcessor = new MessageProcessor();
		msgProcessor.start();
	}
	
	/**
	 * add a state change for stationName in the xml document
	 * @param stationName	name of the station
	 * @param time	like "2008-03-22_14:33:02"
	 * @param change	like "3to5"
	 */
	protected void addStateChange(String stationName, String time, String change) {
		// each change is saved like "3to5@2008-03-22_14:33:02;"; the length of such a string is 25 
		String changeStr = change + "@" + time + ";";
		
		Element eleXml = xmlResDoc.getRootElement();
		Element eleStartTime = eleXml.element("start-time");
		if (eleStartTime == null) {
			eleStartTime = eleXml.addElement("start-time");
			eleStartTime.setText(time);
		}
		
		List lStations = eleXml.elements("station");
		Element eleStation = null;
		for (int i=0; i<lStations.size(); i++) {
			Element tmpStation = (Element)lStations.get(i);
			if (tmpStation.element("id").getText().equalsIgnoreCase(stationName)) {
				eleStation = tmpStation;
				break;
			}
		}
		if (eleStation == null) {
			eleStation = eleXml.addElement("station");
			Element eleId = eleStation.addElement("id");
			eleId.setText(stationName.toLowerCase());
			Element eleChanges = eleStation.addElement("state-changes");
			Element eleGroup = eleChanges.addElement("group");
			eleGroup.setText(changeStr);
			Element eleCount = eleChanges.addElement("change-count");
			eleCount.setText("1");
		} else {
			Element eleChanges = eleStation.element("state-changes");
			// one group contains at most 20 state changes
			int maxGroupLength = 25*20;
			Element eleGroup = null;
			List lGroups = eleChanges.elements("group");
			for (int i=0; i<lGroups.size(); i++) {
				Element tmpGroup = (Element)lGroups.get(i);
				if (tmpGroup.getText().length() < maxGroupLength) {
					eleGroup = tmpGroup;
					break;
				}
			}
			if (eleGroup == null) {
				eleGroup = eleChanges.addElement("group");
				eleGroup.setText(changeStr);
			} else {
				eleGroup.setText(eleGroup.getText() + changeStr);
			}
			Element eleCount = eleChanges.element("change-count");
			eleCount.setText(Integer.toString(Integer.parseInt(eleCount.getText()) + 1));
		}
		
		if (System.currentTimeMillis() - lastSaveTime >= SAVE_INTERVAL)
			saveXmlDoc();
	}
	
	/**
	 * write the xml document to file
	 */
	protected void saveXmlDoc() {					
		String tmpPath = xmlResPath.substring(0, xmlResPath.lastIndexOf('.')) + "_tmp.xml";
		try {
			OutputFormat format = OutputFormat.createPrettyPrint();
			FileWriter fw = new FileWriter(tmpPath);
			XMLWriter writer = new XMLWriter(fw, format);
			writer.write(xmlResDoc);
			writer.close();
			fw.close();
			
			File destFile = new File(xmlResPath);
			File tmpFile = new File(tmpPath); 
			if (destFile.exists()) {
				if (destFile.delete()) {
					tmpFile.renameTo(destFile);					
				}
			} else {
				tmpFile.renameTo(destFile);
			}
			
			lastSaveTime = System.currentTimeMillis();			
			if (System.currentTimeMillis() - lastCreatTime >= CREATE_INTERVAL) {
				xmlResDoc = DocumentHelper.createDocument();
				Element eleXml = xmlResDoc.addElement("xml");
				lastCreatTime = System.currentTimeMillis();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}		
	}
	
	/**
	 * inherited from Filter, for receiving NaradaBrokering events
	 */
	public void onEvent(NBEvent nbEvent) {
		if (nbEvent.getContentPayload() != null) {
			synchronized (listCV) {
				eventList.add(nbEvent);
				if (eventList.size() >= 1) {
					listCV.notify();
				}
			}
		}
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		if (args.length < 5) {
			System.out.println("Usage: java cgl.sensorgrid.sopac.gps.filters.RealtimeRdahmmResulter " +
								"commType nbHost nbPort subTopic xmlResPath");
			System.exit(0);
		}

		String commType = args[0];
		String nbHost = args[1];
		String nbPort = args[2];
		String subTopic = args[3];
		String pubTopic = "";
		String xmlResPath = args[4];
		RealtimeRdahmmResulter test = new RealtimeRdahmmResulter(commType, nbHost, nbPort, subTopic, pubTopic, xmlResPath);
	}

}
